package org.wenzi.com;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;


/**
 * @author zhaozuowen
 * @date 2021-10-18 11:42
 */
public class StreamSqlWordCount {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> source = env.socketTextStream("127.0.0.1", 9000, "\n");

        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        SingleOutputStreamOperator<WordCount> flatMap = source.flatMap(new FlatMapFunction<String, WordCount>() {
            @Override
            public void flatMap(String value, Collector<WordCount> out) throws Exception {
                String[] tokens = value.toLowerCase().split("\\W+");
                for (String token : tokens) {
                    if (token.length() > 0) {
                        out.collect(new WordCount(token, 1));
                    }
                }
            }
        });
        Table table = tableEnvironment.fromDataStream(flatMap,"word,frequency");
        table.printSchema();

        tableEnvironment.createTemporaryView("WordCount",table);

        Table table02 = tableEnvironment.sqlQuery("select word as word ,sum(frequency) as frequency from WordCount group by word");
        DataStream<Tuple2<Boolean, WordCount>> wordCountDataStream = tableEnvironment.toRetractStream(table02, TypeExtractor.createTypeInfo(WordCount.class));
        wordCountDataStream.print();


    }


     public  static class WordCount{
        public String word;
         public int  frequency;

         public WordCount() {
         }

         public WordCount(String word, int  frequency) {
             this.word = word;
             this.frequency = frequency;
         }

         public String getWord() {
             return word;
         }

         public void setWord(String word) {
             this.word = word;
         }

         public Integer getFrequency() {
             return frequency;
         }

         public void setFrequency(int  frequency) {
             this.frequency = frequency;
         }

         @Override
         public String toString() {
             return "WordCount{" +
                     "word='" + word + '\'' +
                     ", frequency=" + frequency +
                     '}';
         }
     }


}
